package http

import (
	"gitee.com/liukexing/mqbroker/component/model"
	"gitee.com/liukexing/mqbroker/core/driver/component"
	"gitee.com/liukexing/mqbroker/core/driver/config"
	"gitee.com/liukexing/mqbroker/core/driver/logger"
	"gitee.com/liukexing/mqbroker/core/server/pubsub"
	"gitee.com/liukexing/mqbroker/core/typedef"
	"github.com/robfig/cron"
	"io/ioutil"
	"net/http"
	"strconv"
	"sync"
	"time"
)

const (
	ConfigName = "config.core.server.http"
)

var (
	Config   typedef.ConfigInfo
	httpMaps map[int]*http.Server
	httpcache sync.Map
	Task   *cron.Cron
	LastTick time.Time
)

func init() {
	var ok bool
	Config, ok = config.GetConfig(ConfigName)
	if !ok {
		return
	}
	LastTick = time.Now().Add(-5 * time.Minute)
}


func Run() bool {
	if !Config.Info.Enable {
		return true
	}
	httpMaps = make(map[int]*http.Server)
	if pubsub.CommandMsgSubServer(Config.Info, NewCommandHandle()) {
		return true
	}
	if pubsub.CommandMsgSubClient(Config.Info, CommandMsgReceiver) {
		return true
	}

	if pubsub.ComputeMsgSubServer(Config.Info, NewComputeHandle()) {
		return true
	}
	if pubsub.ComputeMsgSubClient(Config.Info, ComputeMsgReceiver) {
		return true
	}

	if pubsub.MessageReplyServer(Config.Info, NewReplyHandle()) {
		return true
	}

	initMsgHandle()
	component.InitTask(Config.Info,Task,GetTaskInfo())
	return initHttpServer()
}

func initHttpServer() bool {
	var port []int
	if logger.CheckErr(typedef.ErrInfo{
		Err:  typedef.Decode(Config.Extend["port"], &port),
		Info: Config.Info,
	}) {
		return true
	}
	http.HandleFunc("/", IndexHandler)
	for _, v := range port {
		go startHttpServer(v)
	}
	return false
}

func startHttpServer(p int) {
	logger.Info(typedef.CommInfo{
		Message: "Start HttpServer Port is " + strconv.Itoa(p),
		Info:    Config.Info,
	})
	srv := &http.Server{Addr: ":" + strconv.Itoa(p)}
	httpMaps[p] = srv
	err := srv.ListenAndServe()
	if logger.CheckErr(typedef.ErrInfo{Err: err, Info: Config.Info}) {
		delete(httpMaps, p)
	}
}

func StopHttpServer(p int) {
	logger.Error(typedef.ErrInfo{
		Err:  httpMaps[p].Shutdown(nil),
		Info: Config.Info,
	})
}

func IndexHandler(w http.ResponseWriter, r *http.Request) {
	reqid := typedef.GetUUID()
	body, _ := ioutil.ReadAll(r.Body)
	req := model.HttpRequestData{
		Method:r.Method,
		Proto:r.Proto,
		Header:r.Header,
		Body:body,
		ContentLength:r.ContentLength,
		TransferEncoding:r.TransferEncoding,
		Host:r.Host,
		Path:r.URL.Path,
		RawQuery: r.URL.RawQuery,
		Cookies:r.Cookies(),
	}
	var timeout time.Duration
	if err := typedef.Decode(Config.Extend["timeout"],&timeout);err != nil {
		timeout = 120
	}
	wait := sync.WaitGroup{}
	wait.Add(1)

	rsp := make(chan model.HttpReplyData,1)
	httpcache.Store(reqid,model.HttpWait{
		Data:req,
		Reply:rsp,
		Send:&sync.Once{},
		TimeOut:time.Now().Add(timeout*time.Second),
		Wait:&wait,
	})
	wait.Wait()
	if len(rsp) == 0 {
		dd,_ := httpcache.Load(reqid)
		ChanReply(dd.(model.HttpWait),model.HttpReplyData{
			StatusCode:503,
			Content:[]byte("No Message Response"),
		})
	}
	response := <-rsp
	MsgReply(w,response)
	httpcache.Delete(reqid)
}

func MsgReply(w http.ResponseWriter,reply model.HttpReplyData)  {
	w.WriteHeader(reply.StatusCode)
	for k,v := range reply.Header {
		for _,vs := range  v {
			w.Header().Add(k,vs)
		}
	}
	for _,v:=range reply.Cookies{
		http.SetCookie(w,v)
	}
	w.Write(reply.Content)
}
